package com.aboo.frame.config.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Slf4j
@Component
public class MqConsumer implements ApplicationListener<ContextRefreshedEvent> {

    @Value("${rocketmq.namesrv.addr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.group-name}")
    private String groupName;

    private MqConsumer() {

    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // 避免初始化两次的现象发生
        if (event.getApplicationContext().getParent() == null) {
            try {
                log.info("开始加载RocketMQ消费者，namesrvAddr:{}，groupName:{}", namesrvAddr, groupName);
                // 设置消费者组
                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);

                // BROADCASTING广播模式,CLUSTERING集群模式
                consumer.setMessageModel(MessageModel.CLUSTERING);
                consumer.setVipChannelEnabled(false);
                consumer.setNamesrvAddr(namesrvAddr);
                // 设置消费者端消息拉取策略，表示从哪里开始消费
                // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
                // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
                // CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                // 单次拉取的消息数量
                consumer.setConsumeMessageBatchMaxSize(10);
                // 设置消费者拉取消息的策略，*表示消费该topic下的所有消息，也可以指定tag进行消息过滤
                consumer.subscribe("yata_topic", "*");
                // 消费者端启动消息监听，一旦生产者发送消息被监听到，就打印消息，和rabbitmq中的handlerDelivery类似
                consumer.registerMessageListener(new MqMessageListener());

                // 调用start()方法启动consumer
                consumer.start();
                log.info("加载RocketMQ消费者完成");
            } catch (Exception e) {
                log.error("初始化MQ消费者出错", e);
            }
        }
    }

    public boolean doBusiness(MessageExt messageExt) throws UnsupportedEncodingException {
        String topic = messageExt.getTopic();
        String tag = messageExt.getTags();
        String msg = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
        log.info("*********************************");
        log.info("消费响应  msgId:" + messageExt.getMsgId() + ",msg:" + msg + ",tag:" + tag + ",topic:" + topic);
        log.info("*********************************");

        return true;
    }

    class MqMessageListener implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            ConsumeConcurrentlyStatus status = null;
            try {
                for (MessageExt messageExt : msgs) {
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    boolean bizFlag = doBusiness(messageExt);

                    if (bizFlag) {
                        status = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } else if (reconsumeTimes >= 2) {
                        status = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } else {
                        status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            } catch (Exception e) {
                log.error("MQ消费出错", e);
                status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

            return status;
        }

    }

}
